package net.qiqbframework.modelling.repository;

import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.config.ConfigurationException;
import net.qiqbframework.loadhanding.AggregateLoadBus;
import net.qiqbframework.loadhanding.BatchGenericLoadMessage;
import net.qiqbframework.loadhanding.JdbcAggregate;
import net.qiqbframework.modelling.domain.AggregateModel;
import net.qiqbframework.persisthanding.AggregatePersistBus;
import net.qiqbframework.persisthanding.GenericPersistMessage;
import net.qiqbframework.serialization.Serializer;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;

import static net.qiqbframework.common.BuilderUtils.assertNonNull;

/**
 * 通过 hanlder 来获取或者聚合对象
 *
 * @param <T>
 */
@Slf4j
public class HandlerAggregateRepository<T> extends AbstractAggregateRepository<T, JdbcAggregate<T>> {


    protected final AggregateLoadBus aggregateLoadBus;

    protected final AggregatePersistBus aggregatePersistBus;

    protected final Serializer aggregateRootSerializer;

    protected HandlerAggregateRepository(HandlerAggregateRepository.Builder<T> builder) {
        super(builder);
        this.aggregateLoadBus = builder.aggregateLoadBus;
        this.aggregatePersistBus = builder.aggregatePersistBus;
        this.aggregateRootSerializer = builder.aggregateRootSerializer;
    }

    @Override
    protected JdbcAggregate<T> doCreateNew(Callable<T> factoryMethod) throws Exception {

        final JdbcAggregate<T> initialize = JdbcAggregate.initializeCreate(factoryMethod, aggregateModel(), eventStorageEngine, aggregateRootSerializer);
        return initialize;
    }


    @Override
    public List<JdbcAggregate<T>> doLoads(@Nonnull List<AggregateBizIdentifierVoucher> bizIdentifierVouchers, AggregateModel<T> aggregateModel) {
        // 相同名称的ID分组
        final Map<String, List<AggregateBizIdentifierVoucher>> group = bizIdentifierVouchers.stream()
                .filter(v -> v.getBizIdentifier() != null)
                .collect(Collectors.groupingBy(AggregateBizIdentifierVoucher::getBizIdentifierName));

        JdbcAggregate<T>[] aggregateArray = new JdbcAggregate[bizIdentifierVouchers.size()];
        for (Map.Entry<String, List<AggregateBizIdentifierVoucher>> entry : group.entrySet()) {
            final String bizIdName = entry.getKey();

            aggregateLoadBus.load(new BatchGenericLoadMessage<>(entry.getValue().stream().map(AggregateBizIdentifierVoucher::getBizIdentifier).toList(), bizIdName, aggregateModel), (commandMessage, loadResultMessage) -> {
                if (loadResultMessage.isExceptional()) {
                    throw new RuntimeException("加载异常" + loadResultMessage);
                }
                if (loadResultMessage.getPayload() == null) {
                    return;
                }
                if (loadResultMessage.getPayload() instanceof Map<?, ?> loadResult) {
                    for (Map.Entry<?, ?> resultEntry : loadResult.entrySet()) {
                        final Object bizId = resultEntry.getKey();
                        for (AggregateBizIdentifierVoucher matched : entry.getValue().stream().filter(v -> v.getBizIdentifier().equals(bizId)).toList()) {
                            if (resultEntry.getValue() == null) {
                                aggregateArray[bizIdentifierVouchers.indexOf(matched)] = null;
                            } else {
                                final T aggregateRoot = (T) resultEntry.getValue();
                                final JdbcAggregate<T> jdbcAggregate;
                                try {
                                    jdbcAggregate = JdbcAggregate.initializeLoad(() -> aggregateRoot, aggregateModel, eventStorageEngine, aggregateRootSerializer);
                                } catch (Exception e) {
                                    if (e instanceof RuntimeException runtimeException){
                                        throw runtimeException;
                                    }
                                    throw new RuntimeException(e);
                                }
                                aggregateArray[bizIdentifierVouchers.indexOf(matched)] = jdbcAggregate;
                            }

                        }
                    }

                    return;
                }
                throw new ConfigurationException("加载结果目前只支持map类型");
            });
        }
        // 如果加载凭证为空，且前面有只，则复制到下一个
        for (int i = 1; i < aggregateArray.length; i++) {
            final AggregateBizIdentifierVoucher bizIdentifierVoucher = bizIdentifierVouchers.get(i);
            if (bizIdentifierVoucher.getBizIdentifier() == null) {
                aggregateArray[i] = aggregateArray[i - 1];
            }
        }
        return Arrays.stream(aggregateArray).toList();
    }

    @Override
    protected void doSaves(List<JdbcAggregate<T>> aggregates) {
        if (aggregates == null || aggregates.isEmpty()) {
            return;
        }
        List<GenericPersistMessage<T>> persistMessages = new ArrayList<>();
        for (final JdbcAggregate<T> aggregate : aggregates) {
            GenericPersistMessage<T> persistMessage = new GenericPersistMessage<>(aggregate, Map.of("aggregateRootSnapshot", aggregate.getSnapshotAggregateRoot()));
            persistMessages.add(persistMessage);
        }
        // 填写相同持久化对象
        generatePersistMessage(persistMessages);
        for (GenericPersistMessage<T> persistMessage : persistMessages) {
            try {
                aggregatePersistBus.save(persistMessage);
            } catch (Exception e) {
                if (e instanceof RuntimeException runtimeException){
                    throw runtimeException;
                }
                throw new RuntimeException(e);
            }
        }


    }

    private void generatePersistMessage(List<GenericPersistMessage<T>> persistMessages) {

        for (int i = 0; i < persistMessages.size() - 1; i++) {
            persistMessages.get(i).next(persistMessages.get(i + 1));
        }

    }


    public static <T> Builder<T> builder(AggregateModel<T> aggregateModel) {
        return new Builder<>(aggregateModel);
    }


    public static class Builder<T> extends AbstractAggregateRepository.Builder<T> {

        protected AggregateLoadBus aggregateLoadBus;

        protected AggregatePersistBus aggregatePersistBus;


        protected Serializer aggregateRootSerializer;

        protected Builder(AggregateModel<T> aggregateModel) {
            super(aggregateModel);
        }


        public Builder<T> aggregateLoadBus(@Nonnull AggregateLoadBus aggregateLoadBus) {
            assertNonNull(aggregateLoadBus, "AggregateLoadBus may not be null");
            this.aggregateLoadBus = aggregateLoadBus;
            return this;
        }

        public Builder<T> aggregatePersistBus(@Nonnull AggregatePersistBus aggregatePersistBus) {
            assertNonNull(aggregatePersistBus, "AggregatePersistBus may not be null");
            this.aggregatePersistBus = aggregatePersistBus;
            return this;
        }

        public Builder<T> aggregateRootSerializer(@Nonnull Serializer serializer) {
            assertNonNull(serializer, "serializer may not be null");
            this.aggregateRootSerializer = serializer;
            return this;
        }

        public HandlerAggregateRepository<T> build() {
            return new HandlerAggregateRepository<>(this);
        }
    }
}
